
#include "rtmp_process.h"
#include "rtmp_message.h"
#include "rtmp_resource.h"
#include "rtmp_connection.h"
#include <memory>

RtmpMsgProceess::RtmpMsgProceess()
{
    m_resource_manager = RtmpMediaResourceMgr::Instance();
}

RtmpMsgProceess::~RtmpMsgProceess()
{
}

void RtmpMsgProceess::Process(std::shared_ptr<RtmpConnection> conn, std::shared_ptr<RtmpMessage> & req)
{

    switch(req->m_header.m_type_id)
    {
        case 0x01:
        {
            break;
        }
        case 0x14:
        {
            ProcessCommand(conn, req);
            break;
        }
        case 0x12:
        {
            SetDataFrame(conn, req);
            break;
        }
        case 0x08:
        {
            AudioFrame(conn, req);
            break;
        }
        case 0x09:
        {
            VideoFrame(conn, req);
            break;
        }
    }
}

bool RtmpMsgProceess::CheckConnPower(std::shared_ptr<RtmpConnection> conn)
{
    auto resource = conn->Resource();
    if (resource)
    {
        return false;
    }
    return resource->CheckPushConn(conn);
}

void RtmpMsgProceess::ProcessCommand(std::shared_ptr<RtmpConnection> conn, std::shared_ptr<RtmpMessage> & req)
{
    std::string command = req->CommandString();
    TRACE("process command : {}", command);
    if (command == "connect")
    {
        ConnectCommand(conn, req);
    }
    else if (command == "releaseStream")
    {
        ReleaseStreamCommand(conn, req);
    }
    else if (command == "FCPublish")
    {
        FCPublishCommand(conn, req);
    }
    else if (command == "createStream")
    {
        CreateStreamCommand(conn, req);
    }
    else if (command == "publish")
    {
        PublishCommand(conn, req);
    }
    else if (command == "play")
    {
		PlayStreamCommand(conn, req);
    }
	else if (command == "getStreamLength")
	{
		GetStreamLength(conn,req);
	}
    else if (command == "FCUnpublish")
    {
    
    }
    else if (command == "deleteStream")
    {
    
    }
    // client
    else if (command == "onStatus")
    {
        OnStatusCommand(conn, req);
    }
    else if (command == "")
    {
    
    }
}

void RtmpMsgProceess::ConnectCommand(std::shared_ptr<RtmpConnection> conn, std::shared_ptr<RtmpMessage> & msg)
{
    auto object_data = msg->Object();
    std::string apps;
    std::string tc_url;

    std::shared_ptr<RtmpMessage> resp = std::make_shared<RtmpMessage>();
    auto app_value = object_data->ValueByKey("app");
    auto tc_value  = object_data->ValueByKey("tcUrl");
    if (app_value && tc_value)
    {
        {
            RtmpHeader header;
            header.m_format = 1;
            header.m_chunk_stream = msg->m_header.m_chunk_stream;
            header.m_type_id      = msg->m_header.m_type_id;
            resp->CopyRtmpHeader(header);
        }
        auto _result = new AmfString();
        _result->SetString("_result");

        auto number_value = new AmfNumber();
        number_value->SetValue(msg->NumberId());
        
        auto result_object = new AmfObject();
        result_object->SetString("code", "NetConnection.Connect.Sucess");

        auto result_object_data = new AmfObject();
        result_object_data->SetString("version", "3,5,1,525");

        result_object->SetObject("data", result_object_data);

        result_object->SetString("level", "status");
        result_object->SetNumber("objectEncoding", 0);

        resp->PushAmfValue(_result);
        resp->PushAmfValue(number_value);
        resp->PushAmfValue(result_object);
        {
        
            auto result_sub_object = new AmfObject();
            result_sub_object->SetNumber("capabilities",31);

            result_sub_object->SetString("fmsVer", "FMS/3,5,1,525");
            result_sub_object->SetNumber("mode", 1);

            resp->PushAmfValue(result_sub_object);
        }
        conn->SendRespMsg(resp);
    }        
}

void RtmpMsgProceess::ReleaseStreamCommand(std::shared_ptr<RtmpConnection> conn, std::shared_ptr<RtmpMessage> & msg)
{
    std::shared_ptr<RtmpMessage> resp = std::make_shared<RtmpMessage>();
    {
        RtmpHeader header;
        header.m_format = 1;
        header.m_chunk_stream = msg->m_header.m_chunk_stream;
        header.m_type_id      = msg->m_header.m_type_id;
        resp->CopyRtmpHeader(header);
    }
    auto _result = new AmfString();
    _result->SetString("_result");

    resp->PushAmfValue(_result);

    auto number_value = new AmfNumber();
    number_value->SetValue(msg->NumberId());

    auto null_value = new AmfNull();
    
    resp->PushAmfValue(number_value);
    resp->PushAmfValue(null_value);

    conn->ReleaseStream();
    conn->SendRespMsg(resp);
}

void RtmpMsgProceess::FCPublishCommand(std::shared_ptr<RtmpConnection> conn, std::shared_ptr<RtmpMessage> & msg)
{
    auto resp = std::make_shared<RtmpMessage>();
    {
        RtmpHeader header;
        header.m_format = 1;
        header.m_chunk_stream = msg->m_header.m_chunk_stream;
        header.m_type_id      = msg->m_header.m_type_id;
        resp->CopyRtmpHeader(header);
    }
    auto _result = new AmfString();
    _result->SetString("_result");

    auto number_value = new AmfNumber();
    number_value->SetValue(msg->NumberId());

    auto null_value = new AmfNull();

    resp->PushAmfValue(_result);
    resp->PushAmfValue(number_value);
    resp->PushAmfValue(null_value);

    conn->SendRespMsg(resp);
}

void RtmpMsgProceess::CreateStreamCommand(std::shared_ptr<RtmpConnection> conn, std::shared_ptr<RtmpMessage> & msg)
{
    auto resp = std::make_shared<RtmpMessage>();
    {
        RtmpHeader header;
        header.m_format = 1;
        header.m_chunk_stream = msg->m_header.m_chunk_stream;
        header.m_type_id      = msg->m_header.m_type_id;
        resp->CopyRtmpHeader(header);
    }
    auto _result = new AmfString();
    _result->SetString("_result");

    auto number_value = new AmfNumber();
    number_value->SetValue(msg->NumberId());

    auto null_value = new AmfNull();

    resp->PushAmfValue(_result);
    resp->PushAmfValue(number_value);
    resp->PushAmfValue(null_value);

    conn->SendRespMsg(resp);
}

void RtmpMsgProceess::PublishCommand(std::shared_ptr<RtmpConnection> conn, std::shared_ptr<RtmpMessage> & msg)
{
    auto user_pass_app = msg->UserPass();
    auto user_pass_stream = msg->UserPass(2);
    TRACE("stream name : {} and {}", user_pass_app, user_pass_stream);

    std::string stream_name = "app/" + user_pass_app;
    auto resource = m_resource_manager->Source(stream_name);
    conn->Resource(resource);

    std::shared_ptr<RtmpMessage> publish_result = std::make_shared<RtmpMessage>();
    {
        RtmpHeader header;
        header.m_format = 1;
        header.m_chunk_stream = msg->m_header.m_chunk_stream;
        header.m_type_id      = msg->m_header.m_type_id;
        publish_result->CopyRtmpHeader(header);
    }
    std::string status = "NetStream.Publish.Start";
    std::string description = "Started publishing stream.";

    AmfString * publish_value = new AmfString();
    publish_value->SetString("onFCPublish");
    AmfNumber * number_value  = new AmfNumber();
    number_value->SetValue(0);
    
    AmfObject * publish_object = new AmfObject();
    publish_object->SetString("code", status);
    publish_object->SetString("description", description);

    publish_result->PushAmfValue(publish_value);
    publish_result->PushAmfValue(number_value);
    publish_result->PushAmfValue(publish_object);

    resource->TakePushConn(conn);
    conn->SendRespMsg(publish_result);

    publish_result->m_body.m_amf_value_vec.clear();

    publish_result->PushAmfValue(publish_value);
    publish_result->PushAmfValue(number_value);
    publish_result->PushAmfValue(new AmfNull());
    publish_result->PushAmfValue(publish_object);
    publish_value->SetString("onStatus");
    conn->SendRespMsg(publish_result);
}

void RtmpMsgProceess::SetDataFrame(std::shared_ptr<RtmpConnection> conn, std::shared_ptr<RtmpMessage> & msg)
{
    TRACE("process data meta frame and RtmpSampleAccess");
    std::string command = msg->CommandString();
    if (command == "onMetaData")
    {
        auto resource = conn->Resource();
        if (resource)
        {
            // save meta data frame
            resource->SaveMetadData(msg);
            resource->BroadcastMsg(msg);
        }
    }
    else if (command == "|RtmpSampleAccess")
    {
        auto & data_vec = msg->m_body.m_amf_value_vec;
        for (auto & data : data_vec)
        {
            std::string content = data->AmfValueString();
            INFO(content);
        }
    }
}

void RtmpMsgProceess::AudioFrame(std::shared_ptr<RtmpConnection> conn, std::shared_ptr<RtmpMessage> & msg)
{
    TRACE("process audio frame");
	TRACE("audio timestamp : {}", msg->m_header.m_calc_timestamp);
    auto resource = conn->Resource();
    if (resource)
    {
		bool check_power = resource->CheckPushConn(conn);
		if (check_power)
		{
			resource->BroadcastMsg(msg);
		}

    }
}

void RtmpMsgProceess::VideoFrame(std::shared_ptr<RtmpConnection> conn, std::shared_ptr<RtmpMessage> & msg)
{
    TRACE("process video frame");
	TRACE("video timestamp : {}", msg->m_header.m_calc_timestamp);
    auto resource = conn->Resource();
    if (resource)
    {
		bool check_power = resource->CheckPushConn(conn);
		if (check_power)
		{
			resource->BroadcastMsg(msg);
		}

    }
}

void RtmpMsgProceess::OnStatusCommand(std::shared_ptr<RtmpConnection> conn, std::shared_ptr<RtmpMessage> & msg)
{
    INFO("process onStatus");
    auto & msg_vec = msg->m_body.m_amf_value_vec;
    for (auto & data : msg_vec)
    {
        std::string content = data->AmfValueString();
        INFO(content);
    }
}

void RtmpMsgProceess::SetChunkSize(std::shared_ptr<RtmpConnection> conn, std::shared_ptr<RtmpMessage> & msg)
{

    auto resp = std::make_shared<RtmpMessage>();
    resp->m_header = msg->m_header;

}

void RtmpMsgProceess::PlayStreamCommand(std::shared_ptr<RtmpConnection> conn, std::shared_ptr<RtmpMessage> & msg)
{
	// conn->ReleaseStream();
	INFO("stream type : {}", msg->UserPass());
	// default application for play command
	static const std::string app = "app";


	// command OnStatus NetStream.Play.Reset
	auto play_reset = std::make_shared<RtmpMessage>();
	play_reset->m_header.m_format = 0;
	play_reset->m_header.m_chunk_stream = 5;
	play_reset->m_header.m_timestamp    = 0;
	play_reset->m_header.m_type_id      = 0x14;
	play_reset->m_header.m_stream_id    = 1;

	// play.reset body data
	{
		auto cmd = new AmfString();
		cmd->SetString("onStatus");

		auto number = new AmfNumber();
		number->SetValue(0);

		auto null_data = new AmfNull();

		auto object = new AmfObject();
		{
			object->SetString("level", "status");
			object->SetString("code", "NetStream.Play.Reset");
			object->SetString("description", "Playing and resetting stream.");
			object->SetString("details", "stream");
			object->SetString("clientid", "ASAICiss");

		}
		play_reset->PushAmfValue(cmd);
		play_reset->PushAmfValue(number);
		play_reset->PushAmfValue(null_data);
		play_reset->PushAmfValue(object);

		conn->SendRespMsg(play_reset);
	}
	// command OnStatus NetStream.Play.Star
	auto play_start = std::make_shared<RtmpMessage>();
	play_start->m_header.m_format = 0;
	play_start->m_header.m_chunk_stream = 5;
	play_start->m_header.m_timestamp    = 0;
	play_start->m_header.m_body_size    = 0;
	play_start->m_header.m_type_id      = 0x14;
	play_start->m_header.m_stream_id    = 1;
	{
		auto cmd = new AmfString();
		cmd->SetString("onStatus");

		auto number = new AmfNumber();
		number->SetValue(0);

		auto null = new AmfNull();

		auto object = new AmfObject();
		{
			object->SetString("level", "status");
			object->SetString("code", "NetStream.Play.Start");
			object->SetString("description", "Started playing stream.");
			object->SetString("details", "stream");
			object->SetString("clientid", "ASAICiss");
		}
		play_start->PushAmfValue(cmd);
		play_start->PushAmfValue(number);
		play_start->PushAmfValue(null);
		play_start->PushAmfValue(object);

		conn->SendRespMsg(play_start);
	}
}

void RtmpMsgProceess::GetStreamLength(std::shared_ptr<RtmpConnection> conn, std::shared_ptr<RtmpMessage> & msg)
{
	INFO("stream name is : {}", msg->UserPass());
	static const std::string app = "app";
	std::string stream_name = app + "/" + msg->UserPass();
	INFO("stram is : {}", stream_name);
	auto resource = m_resource_manager->Source(stream_name);
	resource->AddPullConn(conn);
	conn->Resource(resource);  	
}


